package io.github.talelin.sleeve.module.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Component
public class KafkaProducer {

    @Resource
    private KafkaTemplate kafkaTemplate;

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    /**
     * Kafka主题
     */
    @Value("${spring.kafka.topics.topic1}")
    private String topic;

    public void pushKafkaTest(String message) {

        try {
            kafkaTemplate.send(topic, message).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {

                @Override
                public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                    System.out.println("kafka发送成功");
                }

                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("kafka发送失败");
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

